package com.dqlp.yf.rabbitmq.sender;

import com.dqlp.yf.rabbitmq.constant.MqConstant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * @program: rabbitmq
 * @description: mq生产者
 * @author: andy
 * @create: 2019-09-07 10:19
 */
@Component
@Slf4j
public class Sender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    final RabbitTemplate amqpTemplate;


    @Autowired
    public Sender(RabbitTemplate amqpTemplate) {
        amqpTemplate.setConfirmCallback(this::confirm);
        amqpTemplate.setReturnCallback(this::returnedMessage);
        this.amqpTemplate = amqpTemplate;
    }


    public void send(String msg,String routingKey) {
        for (int i = 0; i <= 5; i++) {
            Map<String, Object> map = new HashMap<>();
            map.put("num", i);
            MessageProperties properties = new MessageProperties();
            properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT); //设置持久化
            properties.setHeader("num",i);
          //  properties.setPriority(i); //设置优先级
            Message message = new Message(msg.getBytes(), properties);
            amqpTemplate.convertAndSend(MqConstant.TEST_EXCHANGE, routingKey, message);
        }

    }
    /**
     * 当消息发送到交换机（exchange）时，该方法被调用.
     * 1.如果消息没有到exchange,则 ack=false
     * 2.如果消息到达exchange,则 ack=true
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("*************");
        log.info("correlationData: " + correlationData);
        log.info("ack: " + ack);

        if (!ack) {
            log.info("异常处理....");
        }
        log.info("cause: " + cause);
    }

    /**
     * 当消息从交换机到队列失败时，该方法被调用。（若成功，则不调用）
     * 需要注意的是：该方法调用后，MsgSendConfirmCallBack中的confirm方法也会被调用，且ack = true
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode,
                                String replyText, String exchange, String routingKey) {
        log.info("return exchange: {}, routingKey: {}, replyCode: {}, replyText: {}",
                exchange, routingKey, replyCode, replyText);
        log.info("message: " + new String(message.getBody()));
    }
}
